package com.bbx.flink.table_api_or_sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * DDL    watermark demo    通过ddl eventTime  watermark
 *
 * Database  Sql
 * DB  name :user_active
 * create table ddl
 *  CREATE TABLE `user_active` (
 *   `id` varchar(20) DEFAULT NULL,
 *   `name` varchar(20) DEFAULT NULL,
 *   `active_time` timestamp(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3)
 *   ) ENGINE=InnoDB DEFAULT CHARSET=utf8
 */
public class TimeAttributeDDLEventTimeDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String createSql = "CREATE TABLE userActive (\n" +
                "id String,    \n" +
                "name String,  \n" +
                "active_time TIMESTAMP(3)  , \n" +
                "WATERMARK FOR active_time AS active_time - INTERVAL '5' SECOND \n  " +
                ") WITH (\n " +
                "  'connector.type' = 'jdbc',\n" +
                "  'connector.url' = 'jdbc:mysql://192.168.10.7:3306/flink_user?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC',\n" +
                "  'connector.table' = 'user_active', \n" +
                "  'connector.driver' = 'com.mysql.cj.jdbc.Driver',\n" +
                "  'connector.username' = 'root',\n" +
                "  'connector.password' = '123456')" ;

        tableEnv.executeSql(createSql);
//        String querySql = "select * from  userActive ";
        String querySql = "SELECT TUMBLE_START(active_time, INTERVAL '10' MINUTE), COUNT(DISTINCT name)\n" +
                "FROM userActive GROUP BY TUMBLE(active_time, INTERVAL '10' MINUTE)  ";
        tableEnv.executeSql(querySql).print();
    }


}
